package com.yanyu.space.common.log.sender.kafka;

import com.yanyu.space.common.core.properties.LogProperties;
import com.yanyu.space.common.log.message.LogMessage;
import com.yanyu.space.common.log.sender.LogSender;
import lombok.SneakyThrows;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.InitializingBean;

import java.util.Properties;

/**
 * @author yanyu
 */
public class KafkaLogSender implements LogSender, InitializingBean {

    private KafkaProducer<String, String> producer;

    private LogProperties kafkaLogProperties;

    public KafkaLogSender(LogProperties kafkaLogProperties) {
        this.kafkaLogProperties = kafkaLogProperties;
    }

    @SneakyThrows
    @Override
    public void send(LogMessage logMessage) {
        ProducerRecord<String, String> record = new ProducerRecord(kafkaLogProperties.getSender().getTopic(), logMessage.toString());
        try {
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void afterPropertiesSet() {
        Properties properties = new Properties();
        // 如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 50);
        // 只有数据积累到batch.size之后，sender才会发送数据。默认16k
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 32 * 1024);
        // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 设置可靠性;
        // acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
        // acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
        // acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        // 设置重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 1);
        // 设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置值的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 设置集群地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaLogProperties.getSender().getServersAddr());
        producer = new KafkaProducer<>(properties);
    }
}
